package com.liustar.mq.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author Leo
 */
public class RocketMQConsumer {
    private final Logger logger = Logger.getLogger(this.getClass().getName());
    private String topic;
    private String tags;
    private String keys;
    private String consumerGroup;
    private String namesrvAddr;
    private MessageListener messageListener;

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getTags() {
        return tags;
    }

    public void setTags(String tags) {
        this.tags = tags;
    }

    public String getKeys() {
        return keys;
    }

    public void setKeys(String keys) {
        this.keys = keys;
    }

    public String getConsumerGroup() {
        return consumerGroup;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public String getNamesrvAddr() {
        return namesrvAddr;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public MessageListener getMessageListener() {
        return messageListener;
    }

    public void setMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }

    public void init() {
        logger.log(Level.INFO, "Start RocketMQ consumer listener..." + this.getTopic() + ":" + this.getTags());
        DefaultMQPushConsumer consumer =
                new DefaultMQPushConsumer();
        consumer.setConsumerGroup(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setInstanceName(consumerGroup + System.currentTimeMillis());
        try {
            //订阅PushTopic下Tag为push的消息
            consumer.subscribe(topic, tags);
            //程序第一次启动从消息队列头取数据
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            RocketMqMessageWrapper rocketMqMessageWrapper = new RocketMqMessageWrapper();
            if (this.messageListener == null) {
                throw new RuntimeException("please define a mqMessageListener for consumer!");
            }
            rocketMqMessageWrapper.setMqMessageListener(this.messageListener);
            consumer.registerMessageListener(rocketMqMessageWrapper);
            consumer.start();
            logger.log(Level.INFO, "The RocketMQ consumer listener<" + messageListener.getClass().getName() + "> started successfully！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
